package com.coding4play.mqtt.broker.internal;

import com.coding4play.mqtt.broker.config.BrokerProperties;
import com.coding4play.mqtt.broker.util.JdkTool;
import io.netty.handler.codec.mqtt.MqttQoS;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;


@Component
@AllArgsConstructor
@Slf4j
public class RedisListener implements MessageListener {
    public static final String CLUSTER_TOPIC = "mqtt:cluster:listener";

    private RedisMessageListenerContainer listenerContainer;
    private BrokerProperties brokerProperties;
    private InternalBroadcast internalBroadcast;

    @PostConstruct
    public void postConstruct() {
        this.listenerContainer.addMessageListener(this, PatternTopic.of(CLUSTER_TOPIC));
    }

    @Override
    public void onMessage(Message message, byte[] pattern) {
        Object obj = RedisSerializer.json().deserialize(message.getBody());
        InternalMessage msg = (InternalMessage) obj;
        log.info("redisListener: {}", msg);

        //判断进程ID是否是自身实例,若相同则不发送,否则集群模式下重复发消息
        if ( !brokerProperties.getId().equals(msg.getBrokerId())
                && !JdkTool.getProcessId("0").equals(msg.getProcessId()))
            internalBroadcast.broadcastMessage(
                    msg.getClientId(),
                    msg.getTopic(),
                    MqttQoS.valueOf(msg.getMqttQoS()),
                    msg.getMessageBytes(),
                    msg.getRetain(),
                    msg.getDup()
            );
    }

}
